package cn.doitedu.dwetl.utils

import java.io.{ByteArrayInputStream, ByteArrayOutputStream, DataInputStream, DataOutputStream}

import org.apache.spark.sql.functions.udf
import org.roaringbitmap.RoaringBitmap

import scala.collection.mutable

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-01-17
 * @desc roaringbitmap的序列化和反序列化工具
 */
object RrUtils {

  /**
   * 序列化
   * @param rr
   * @return
   */
  def ser(rr:RoaringBitmap):Array[Byte]={
    val ba = new ByteArrayOutputStream()
    val dout = new DataOutputStream(ba)

    rr.serialize(dout)

    ba.toByteArray
  }


  def de(arr:Array[Byte]):RoaringBitmap={
    val rr = new RoaringBitmap()

    val bainput = new ByteArrayInputStream(arr)
    val dataInput = new DataInputStream(bainput)
    rr.deserialize(dataInput)

    rr
  }


  /**
   * 取bitmap中的1的个数（基数）
   * @param bits
   * @return
   */
  def getCard(bits:Array[Byte]) = {
    val rrbm = new RoaringBitmap()
    val bainput = new ByteArrayInputStream(bits)
    val dataInput = new DataInputStream(bainput)
    rrbm.deserialize(dataInput)

    rrbm.getCardinality
  }


  val toBitmap = (ids:mutable.WrappedArray[Int])=>{

    val bitmap = RoaringBitmap.bitmapOf(ids: _*)

    val bout = new ByteArrayOutputStream()

    // 利用jdk自己的objectoutputstream序列化的效率较低
    /*val objOutput = new ObjectOutputStream(bout)
    objOutput.writeObject(bitmap)*/

    // 利用roaringbitmap自带的序列化方法，效率较高
    val dataOutput = new DataOutputStream(bout)
    bitmap.serialize(dataOutput)

    bout.toByteArray
  }




}
